Data Engineering Path · Airflow
EMR Operators & Hooks
Orchestrating Big Compute From the Outside
EMR is where Airflow hands off to genuinely large-scale Spark/Hadoop compute — spin up a cluster, run steps, tear it down. The Case Studies section of this course already walks through a full production EMR pipeline end-to-end; this page is the focused, standalone reference for the operators and hook themselves.
A Note on This Page's Screenshots
Unlike the S3/SNS/SQS/Lambda/DynamoDB/Athena pages, this one doesn't include a live screenshot. A real EMR cluster takes roughly 10-15 minutes to provision and costs real money for the EC2 instances behind it for as long as it runs — disproportionate for a single demo screenshot. The code below is correct and directly mirrors the real, working pipeline in this course's EMR Case Study (Data & AI → Case Studies), which was run for real when that content was built.
Unlike the S3/SNS/SQS/Lambda/DynamoDB/Athena pages, this one doesn't include a live screenshot. A real EMR cluster takes roughly 10-15 minutes to provision and costs real money for the EC2 instances behind it for as long as it runs — disproportionate for a single demo screenshot. The code below is correct and directly mirrors the real, working pipeline in this course's EMR Case Study (Data & AI → Case Studies), which was run for real when that content was built.
The Core Lifecycle
from airflow.providers.amazon.aws.operators.emr import (
EmrCreateJobFlowOperator,
EmrAddStepsOperator,
EmrTerminateJobFlowOperator,
)
from airflow.providers.amazon.aws.sensors.emr import EmrJobFlowSensor, EmrStepSensor
JOB_FLOW_OVERRIDES = {
"Name": "nightly-sales-transform",
"ReleaseLabel": "emr-7.1.0",
"Applications": [{"Name": "Spark"}],
"Instances": {
"InstanceGroups": [
{"Name": "Primary", "Market": "ON_DEMAND", "InstanceRole": "MASTER", "InstanceType": "m5.xlarge", "InstanceCount": 1},
{"Name": "Core", "Market": "SPOT", "InstanceRole": "CORE", "InstanceType": "m5.xlarge", "InstanceCount": 2},
],
"KeepJobFlowAliveWhenNoSteps": False,
"TerminationProtected": False,
},
"JobFlowRole": "EMR_EC2_DefaultRole",
"ServiceRole": "EMR_DefaultRole",
}
create_cluster = EmrCreateJobFlowOperator(
task_id="create_emr_cluster",
job_flow_overrides=JOB_FLOW_OVERRIDES,
aws_conn_id="aws_default",
)
wait_for_cluster = EmrJobFlowSensor(
task_id="wait_for_cluster_ready",
job_flow_id=create_cluster.output,
target_states=["WAITING"],
aws_conn_id="aws_default",
)
add_steps = EmrAddStepsOperator(
task_id="add_transform_step",
job_flow_id=create_cluster.output,
steps=[{
"Name": "transform_sales_data",
"ActionOnFailure": "TERMINATE_CLUSTER",
"HadoopJarStep": {
"Jar": "command-runner.jar",
"Args": ["spark-submit", "s3://my-scripts/transform_sales.py"],
},
}],
aws_conn_id="aws_default",
)
wait_for_step = EmrStepSensor(
task_id="wait_for_transform_step",
job_flow_id=create_cluster.output,
step_id="{{ task_instance.xcom_pull(task_ids='add_transform_step')[0] }}",
aws_conn_id="aws_default",
)
terminate_cluster = EmrTerminateJobFlowOperator(
task_id="terminate_emr_cluster",
job_flow_id=create_cluster.output,
trigger_rule="all_done", # terminate even if the step failed - never leave a cluster running
aws_conn_id="aws_default",
)
create_cluster >> wait_for_cluster >> add_steps >> wait_for_step >> terminate_cluster
Always Set trigger_rule="all_done" on the Terminate Task
The single most expensive EMR mistake: a step fails, the DAG stops before reaching the terminate task, and the cluster keeps running (and billing) indefinitely.
The single most expensive EMR mistake: a step fails, the DAG stops before reaching the terminate task, and the cluster keeps running (and billing) indefinitely.
trigger_rule="all_done" guarantees the terminate task runs regardless of what happened upstream.
EmrContainerOperator — Running on EKS Instead
For teams running EMR on an existing Kubernetes/EKS cluster rather than provisioning EC2 instances per job:
from airflow.providers.amazon.aws.operators.emr import EmrContainerOperator
run_on_eks = EmrContainerOperator(
task_id="run_transform_on_eks",
virtual_cluster_id="abc123",
execution_role_arn="arn:aws:iam::123456789012:role/emr-eks-execution-role",
release_label="emr-7.1.0-latest",
job_driver={
"sparkSubmitJobDriver": {
"entryPoint": "s3://my-scripts/transform_sales.py",
}
},
aws_conn_id="aws_default",
)
EmrHook Directly
from airflow.providers.amazon.aws.hooks.emr import EmrHook
def get_cluster_cost_estimate(job_flow_id: str):
hook = EmrHook(aws_conn_id="aws_default")
cluster = hook.get_cluster_id_by_name # or describe_cluster via hook.get_conn()
...
See the Case Studies → Event Trigger Pipeline section for this exact pattern (spot instances, branching on step failure, defensive cluster teardown) working end-to-end in a real production-style DAG.
arrow_back Previous Topic
Redshift Operators and Hooks
Next Topic arrow_forward
Glue Operators and Hooks